package skewreduce.framework;

import java.io.File;
import java.io.IOException;
import java.net.URL;
import java.util.Arrays;
import java.util.Properties;

import org.apache.log4j.PropertyConfigurator;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.conf.Configured;
import org.apache.hadoop.fs.FileStatus;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.util.GenericOptionsParser;
import org.apache.hadoop.util.ReflectionUtils;
import org.apache.hadoop.util.Tool;
import org.apache.hadoop.util.ToolRunner;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import skewreduce.framework.logical.LFinalizeOp;
import skewreduce.framework.logical.LMergeOp;
import skewreduce.framework.logical.LPartitionOp;
import skewreduce.framework.logical.LSampleOp;
import skewreduce.framework.logical.LProcessOp;
import skewreduce.lib.Partition;


public abstract class EaggDriver extends Configured implements Tool {
    protected static final Logger LOG = LoggerFactory.getLogger(EaggDriver.class);

    public abstract Partition getRootPartition();
    public abstract String getAppName();
    public abstract LSampleOp    createLSampleOp();
    public abstract LPartitionOp createLPartitionOp();
    public abstract LProcessOp createLWorkOp();
    public abstract LMergeOp createLMergeOp();
    public abstract LFinalizeOp createLFinalizeOp();

    private FileSystem fs;
    public final FileSystem getFileSystem() throws IOException {
        if ( fs == null ) {
            synchronized (this) {
                if ( fs == null ) {
                    fs = FileSystem.get(getConf());
                }
            }
        }
        return fs;
    }

    final String LOG4J_PROP = "log4j.properties";

    private void configureLog() {
        String log4j = getConf().get(LOG_CONFIG_ATTR);
        File f = null;

        if ( log4j == null ) {
            // search logging file in current directory
            f = new File(LOG4J_PROP);
        } else {
            f = new File(log4j);
        }
        if ( f.exists() ) {
            System.err.println("Setting up logging from "+f.getPath());
            PropertyConfigurator.configure(f.getPath());
            return;
        } else {
            // try to load from local resource bundle
            URL url = EaggDriver.class.getResource("/skewreduce/"+LOG4J_PROP);
            if ( url != null ) {
                System.err.println("Setting up logging from default properties");
                PropertyConfigurator.configure(url);
                return;
            }
        }

        System.err.println("Setting up basic logging");

        // configure a console
        Properties props = new Properties();
        props.setProperty("log4j.rootLogger","INFO, CONSOLE");
        props.setProperty("log4j.appender.CONSOLE","org.apache.log4j.ConsoleAppender");
        props.setProperty("log4j.appender.CONSOLE.layout","org.apache.log4j.PatternLayout");
        props.setProperty("log4j.appender.CONSOLE.layout.ConversionPattern","%d [%t] %-5p %c - %m%n");
        props.setProperty("log4j.appender.CONSOLE.target","System.err");

        PropertyConfigurator.configure(props);
    }

    protected int handleOption(String[] args,int i) {
        return -1;
    }

    public final int run(String[] args) throws Exception {
        Path input = null;
        Path output = null;
        boolean bad = false;

        for ( int i = 0; i < args.length; ++i ) {
            if ( "-input".equals(args[i]) ) {
                if ( i+1 < args.length ) {
                    input = new Path(args[++i]);
                } else {
                    System.err.println("missing -input parameter");
                    bad = true;
                }
            } else if ( "-output".equals(args[i]) ) {
                if ( i+1 < args.length ) {
                    output = new Path(args[++i]);
                } else {
                    System.err.println("missing -output parameter");
                    bad = true;
                }
            } else {
                int nargs = handleOption(args,i);
                if ( nargs < 0 ) {
                    System.err.println("Unknown option: "+args[i]);
                    bad = true;
                } else {
                    i += nargs;
                }
            }
        }

        FileSystem fs = getFileSystem();

        if ( input == null || output == null ) {
            System.err.println("Missing input or output");
            bad = true;
        } else {
            Path testOut  = new Path(output,"part-?-*");

            if ( input != null && ! fs.exists(input) ) {
                // check whether it's glob pattern
                FileStatus[] stats = fs.globStatus(input);
                if ( stats.length == 0 ) {
                    System.err.println(input.toUri()+": does not exists");
                    bad = true;
                }
            }
            if ( output != null && fs.globStatus(testOut).length > 0 ) {
                System.err.println(output.toUri()+": already exists");
                bad = true;
            }
        }

        if ( bad ) {
            return 1;
        }

        configureLog();

        // build information
        EaggPlan plan = EaggPlanCompiler.compile(this,input,output);
        
        // create executor
        final EaggExecutor  executor = new EaggExecutor(getConf(),this,plan);
        Runtime.getRuntime().addShutdownHook(new Thread() {
            public void run() { executor.shutdown(); }
        });

        int rc = executor.execute() ? 0 : 1;

        plan.exportDot( getAppName() + "-plan.final.dot");
        
        return rc;
    }
    
    private static final String DRIVER_CLASS_ATTR = "skewreduce.driver.class";
    private static final String LOG_CONFIG_ATTR = "skewreduce.log4j";

    public static void main(String[] args) throws Exception {
        GenericOptionsParser options = new GenericOptionsParser(new Configuration(),args);
        Configuration conf = options.getConfiguration();
        String[] rest = options.getRemainingArgs();

        if ( rest == null || rest.length == 0 ) {
            System.err.println("No class argument is given!");
            System.exit(1);
        }

        if ( conf.get(DRIVER_CLASS_ATTR) == null ) {
            conf.set(DRIVER_CLASS_ATTR,rest[0]);
            System.err.println("Using application "+rest[0]);
            rest = (rest.length > 1 ) ? Arrays.copyOfRange(rest, 1, rest.length) : new String[0];
        }

        Class<? extends EaggDriver> driverClass = conf.getClass(DRIVER_CLASS_ATTR,EaggDriver.class,EaggDriver.class);

        EaggDriver driver = (EaggDriver)ReflectionUtils.newInstance(driverClass,conf);
        int rc = ToolRunner.run(driver,rest);

        System.exit(rc);
    }
}
